#pragma once
#include <algorithm>
#include <typeinfo>
#include <asm-generic/socket.h>
#include <cassert>
#include <cerrno>
#include <cstddef>
#include <cstdint>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <functional>
#include <iostream>
#include <memory>
#include <queue>
#include <sys/time.h>
#include <sys/types.h>
#include <unordered_map>
#include <utility>
#include <vector>
#include <iostream>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <sys/timerfd.h>
#include <thread>
#include <mutex>
#include <sys/eventfd.h>
// ********************************
// 首先实现的是我们的日志宏
// 日志等级的划分
#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL DBG

#define LOG(level, format, ...) do { \
    if(level < LOG_LEVEL) {\
        break; \
    } \
    time_t t = time(NULL); \
    struct tm* ltm = localtime(&t); \
    char tmp[32] = {0}; \
    strftime(tmp, 31, "%H:%M:%S", ltm); \
    fprintf(stdout, "[%p %s %s:%d] " format "\n", (void*)pthread_self(), tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
}while(0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

// ********************************
// 这里实现的是缓冲区模块

#define BUFFER_DEFAULT_SIZE 1024
class Buffer {
private:
    std::vector<char> _buffer;
    uint64_t _read_index; // 读偏移
    uint64_t _write_index; // 写偏移
public:
    Buffer() : _buffer(BUFFER_DEFAULT_SIZE), _read_index(0), _write_index(0) {}
    char* Begin() {
        return &(*(_buffer.begin()));
    }
    char* WritePosition() {
        return Begin() + _write_index;
    }
    char* ReadPosition() {
        return Begin() + _read_index;
    }
    uint64_t TailIdleSize() {
        return _buffer.size() - _write_index;
    }
    uint64_t HeadIdleSize() {
        return _read_index;
    }
    uint64_t ReadAbleSize() {
        return _write_index - _read_index;
    }
    void MoveReadOffset(uint64_t len) {
        assert(len <= ReadAbleSize());
        _read_index += len;
    }
    void MoveWriteOffset(uint64_t len) {
        assert(len <= TailIdleSize());
        _write_index += len;
    }
    void EnsurWriteSpace(uint64_t len) {
        if(len <= TailIdleSize()) {
            return;
        }
        if(len <= TailIdleSize() + HeadIdleSize()) {
            uint64_t sz = ReadAbleSize();
            std::copy(ReadPosition(), WritePosition(), Begin());
            _read_index = 0;
            _write_index = sz;
        }else {
            DBG_LOG("resize:%ld", _write_index + len);
            _buffer.resize(_write_index + len);
        }
    }
    void Write(const void* data, uint64_t len) {
        EnsurWriteSpace(len);
        const char* index = (const char*)data;
        std::copy(index, index + len, WritePosition());
    }
    void WriteAndPush(const void* data, uint64_t len) {
        Write(data, len);
        MoveWriteOffset(len);
    }
    void WriteString(const std::string data) {
        Write(data.c_str(), data.size());
    }
    void WriteStringAndPush(const std::string data) {
        WriteString(data);
        MoveWriteOffset(data.size());
    }
    void WriteBuffer(Buffer& data) {
        Write(data.ReadPosition(), data.ReadAbleSize());
    }
    void WriteBufferAndPush(Buffer& data) {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAbleSize());
    }
    void Read(void* buf, uint64_t len) {
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(), WritePosition(), (char*)buf);
    }
    void ReadAndPop(void* buf, uint64_t len) {
        Read(buf, len);
        MoveReadOffset(len);
    }
    std::string ReadAsString(uint64_t len) {
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }
    std::string ReadAsStringAndPop(uint64_t len) {
        assert(len <= ReadAbleSize());
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }
    char* FindCRLF() {
        char* res = (char*)memchr(ReadPosition(), '\n', ReadAbleSize());
        return res;
    }
    std::string GetLine() {
        char* pos = FindCRLF();
        if(pos == NULL) {
            return "";
        }else {
            return ReadAsString(pos - ReadPosition() + 1);
        }
    }
    std::string GetLinePop() {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }
    void Clear() {
        _read_index = 0;
        _write_index = 0;
    }
};


// ******************************
// 这里实现的是套接字模块
#define BacklogMax 1024
class Socket {
private:
    int _sockfd;
public:
    Socket() : _sockfd(-1) {}
    Socket(int sockfd) : _sockfd(sockfd) {}
    ~Socket() {
        Close();
    }
    void Close() { 
        if(_sockfd != -1) {
            close(_sockfd);
            _sockfd = -1;
        }
    }
    int Fd() {
        return _sockfd;
    }
    // 创建套接字
    bool Create() {
        _sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if(_sockfd < 0) {
            ERR_LOG("socket error!");
            return false;
        }
        return true;
    }
    // 绑定相关信息
    bool Bind(const std::string& ip, uint16_t port) {
        struct sockaddr_in addr;
        int len = sizeof(addr);
        addr.sin_port = htons(port);
        addr.sin_family = AF_INET;
        inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);
        int n = bind(_sockfd, (struct sockaddr*)&addr, len);
        if(n < 0) {
            ERR_LOG("bind error");
            return false;
        }
        return true;
    }
    // 开始监听
    bool Listen(int backlog = BacklogMax) {
        int fd = listen(_sockfd, backlog);
        if(fd < 0) {
            ERR_LOG("listen error");
            return false;
        }
        return true;
    }
    // 向服务器发起连接操作（客户端）
    bool Connect(const std::string& ip, uint16_t port) {
        struct sockaddr_in addr;
        int len = sizeof(addr);
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);
        int n = connect(_sockfd, (struct sockaddr*)&addr, len);
        if(n < 0) {
            ERR_LOG("connect error");
            return false;
        }
        return true;
    }
    // 获取连接
    int Accept() {
        int ret = accept(_sockfd, NULL, NULL);
        if(ret < 0) {
            ERR_LOG("accept error");
            return -1;
        }
        return ret;
    }
    // 接收数据
    size_t Recv(void* buf, int len, int flag = 0) {
        int size = recv(_sockfd, buf, len, flag);
        if(size <= 0) {
            if(errno == EINTR || errno == EAGAIN) // 前面的是被信号给打断了，后面是没有数据了（一般是非阻塞才有的额情况）
            {
                return 0; 
            }
            ERR_LOG("recv error");
            return -1; // 对端关闭返回的是-1，和原来的逻辑是不一样的
        }
        return size;
    }
    size_t NonBlockRecv(void* buf, int len) {
        return Recv(buf, len, MSG_DONTWAIT);
    }
    // 发送数据
    size_t Send(const void* buf, int len, int flag = 0) {
        int size = send(_sockfd, buf, len, flag);
        if(size < 0) {
            if(errno == EINTR || errno == EAGAIN) {
                return 0;
            }
            return -1;
        }
        return size;
    }
    size_t NonBlockSend(void* buf, int len) {
        return Send(buf, len, MSG_DONTWAIT);
    }
    // 创建一个服务端的连接
    bool CreateServer(uint16_t port, const std::string& ip = "0.0.0.0", bool block_log = false) {
        if(Create() == false) {
            return false;
        }
        if(block_log) {
            NonBlock(); // 这里设置监听套接字为非阻塞，可以防止我们accept阻塞，主要还是为了高并发的场景
        }
        if(Bind(ip, port) == false) {
            return false;
        }
        if(Listen() == false) {
            return false;
        }
        ReuseAddress();
        return true;
    }
    // 创建一个客户端连接
    bool CreateClient(uint16_t port, const std::string& ip) {
        if(Create() == false) {
            return false;
        }
        if(Connect(ip, port) == false) {
            return false;
        }
        return true;
    }
    // 设置套接字非阻塞
    void NonBlock() {
        int flag = fcntl(_sockfd, F_GETFL, 0); // 获取当前的标志
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK); // 添加非阻塞标志
    }
    void ReuseAddress() {
        int opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); // 这里设置的是允许绑定到一个已经使用过的本地地址和端口号，常用来快速重启的服务端应用程序
        opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)); // 这里设置的是允许多个套接字绑定到同一个端口，常用于实现负载均衡等高并发的场景
    }
};

// ******************************
// 这里实现的是channel模块,是对一个描述符需要进行的IO事件进行管理的模块
class Poller;
class EventLoop;
class Channel {
private:
    int _fd;
    EventLoop* _loop;
    uint32_t  _events; // 需要监控的事件
    uint32_t _revents; // 当前连接触发的事件
    using EventCallback = std::function<void()>;
    EventCallback _read_callback;
    EventCallback _write_callback;
    EventCallback _error_callback;
    EventCallback _close_callback;
    EventCallback _event_callback;
public:
    Channel(EventLoop* loop, int fd) : _fd(fd), _loop(loop), _events(0), _revents(0) {}
    int Fd() {
        return _fd;
    }
    uint32_t Events() {
        return _events;
    }
    void SetREvent(uint32_t revents) {
        _revents = revents;
    }
    void SetReadCallback(const EventCallback read_callback) {
        _read_callback = read_callback;
    }
    void SetWriteCallback(const EventCallback write_callback) {
        _write_callback = write_callback;
    }
    void SetErrorCallback(const EventCallback error_callback) {
        _error_callback = error_callback;
    }
    void SetCloseCallback(const EventCallback close_callback) {
        _close_callback = close_callback;
    }
    void SetEventCallback(const EventCallback event_callback) {
        _event_callback = event_callback;
    }
    // 监控是否可读
    bool ReadAble() {
        return (_events & EPOLLIN);
    }
    // 监控是否可写
    bool WriteAble() {
        return (_events & EPOLLOUT);
    }
    // 启动读事件监控
    void EnableRead() {
        _events |= EPOLLIN;
        Update();
    }
    // 启动写事件监控
    void EnableWrite() {
        _events |= EPOLLOUT;
        Update();
    }
    // 关闭写事件监控
    void DisableRead() {
        _events &= ~EPOLLIN;
        Update();
    }
    // 关闭读事件监控
    void DisableWrite() {
        _events &= ~EPOLLOUT;
        Update();
    }
    // 关闭所有的事件监控
    void DisableAll() {
        _events = 0;
        Update();
    }
    void Update();
    void Remove();
    // 处理事件，触发事件的时候调用
    void HandleEvent() {
        if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || _revents & EPOLLPRI) { // 第二个是对端退出（写端），第三个是优先数据
            if(_read_callback) {
                _read_callback();
            }
        }
        if(_revents & EPOLLOUT) {
            if(_write_callback) {
                _write_callback();
            }
        }else if(_revents & EPOLLHUP) {
            if(_close_callback) {
                _close_callback();
            }
        }else if(_revents & EPOLLERR) {
            if(_error_callback) {
                _error_callback();
            }
        }
        if(_event_callback) { // 不管什么事件都要调用的事件
            _event_callback();
        }
    }
};
// ******************************
// 这里实现的是poller模块，主要是对我们的epoll进行一个封装
#define Max_EPOLLEVENTS 1024
class Poller {
private:
    int _epfd;
    struct epoll_event _evs[Max_EPOLLEVENTS];
    std::unordered_map<int, Channel*> _channels;
public:
    // 对epoll的直接操作
    void Update(Channel* channel, int op) {
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->Events();
        int n = epoll_ctl(_epfd, op, fd, &ev);
        if(n < 0) {
            ERR_LOG("epollctl error");
        }
        return;
    }
    // 判断是否添加了事件的监控
    bool HasChannel(Channel* channel) {
        auto it = _channels.find(channel->Fd());
        if(it == _channels.end()) {
            return false;
        }
        return true;
    }
public:
    Poller() {
        _epfd = epoll_create(Max_EPOLLEVENTS);
        if(_epfd < 0) {
            ERR_LOG("epollcreate error");
            abort(); // 退出程序，直接退出，啥也没干
        }
    }
    // 添加或是修改事件
    void UpdateEvent(Channel* channel) {
        bool ret = HasChannel(channel);
        // 存在就修改，不存在就添加
        if(!ret) {
            _channels.insert(std::make_pair(channel->Fd(), channel));
            return Update(channel, EPOLL_CTL_ADD);
        }
        return Update(channel, EPOLL_CTL_MOD);
    }
    // 移除监控
    void RemoveEvent(Channel* channel) {
        auto it = _channels.find(channel->Fd());
        if(it != _channels.end()) {
            _channels.erase(it);
        }
        Update(channel, EPOLL_CTL_DEL);
    }
    // 开始监控
    void Poll(std::vector<Channel*> *active) { // 输入输出型参数
        int n = epoll_wait(_epfd, _evs, Max_EPOLLEVENTS, -1);
        if(n < 0) {
            if(errno == EINTR) { // 中断，可以重来
                return;
            }
            // 直接结束，并打印是什么错误
            ERR_LOG("epoll wait error: %s\n", strerror(errno));
            abort();
        }
        for(int i = 0; i < n; i++) {
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());
            it->second->SetREvent(_evs[i].events);
            active->push_back(it->second); // 添加
        }
        return;
    }
};
// ******************************
// 这里实现的是我们定时器任务类
using TaskFunc = std::function<void()>; // 要执行的定时器任务
using ReleaseFunc = std::function<void()>; // 用来删除timerwhell中保存的定时器对象的信息
class TimeTask {
private:
    uint64_t _id; // 定时器任务对象的id;
    uint32_t _timeout; // 定时任务的超时时间
    bool _canceled; // 是否删除了
    TaskFunc _task_cb; // 要执行的定时任务
    ReleaseFunc _release; // 用来删除timerwhell中保存的定时器对象的信息
public:
    TimeTask(uint64_t id, uint32_t delay, const TaskFunc& cb)
    :_id(id), _timeout(delay), _task_cb(cb), _canceled(false) {}
    ~TimeTask() {
        if(_canceled == false) {
            _task_cb();
        }
        _release();
    }
    void Cancel() {
        _canceled = true;
    }
    void SetRelease(const ReleaseFunc& cb) {
        _release = cb;
    }
    uint32_t DelayTime() {
        return _timeout;
    }
};
// ******************************
// 这里实现的是我们的时间轮类
class TimeWheel {
private:
    using WeakTask = std::weak_ptr<TimeTask>;
    using SharedTask = std::shared_ptr<TimeTask>;
    int _tick; // 当前的秒针，走到了哪里就释放哪里（也就是执行相关的任务）
    int _capacity; // 表盘的最大数量，也就是最大的延迟
    std::vector<std::vector<SharedTask>> _wheel; // 时间轮
    std::unordered_map<uint64_t, WeakTask> _timers; // 存储相关的定时器任务的信息，方便判断是否存在
    EventLoop* _loop;
    int _timerfd;
    std::unique_ptr<Channel> _timer_channel;
private:
    void RemoveTimer(uint64_t id) {
        auto it = _timers.find(id);
        if(it != _timers.end()) {
            _timers.erase(it);
        }
        return;
    }
    static int CreateTimefd() {
        int timefd = timerfd_create(CLOCK_MONOTONIC, 0);
        if(timefd < 0) {
            ERR_LOG("timefd create error");
            abort();
        }
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;
        timerfd_settime(timefd, 0, &itime, NULL);
        return timefd;
    }
    int ReadTimefd() {
        uint64_t times;
        int n = read(_timerfd, &times, 8);
        if(n < 0) {
            ERR_LOG("read timefd error");
            abort();
        }
        return times;
    }
    void RunTimerTask() {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear();
    }
    void OnTime() {
        int times = ReadTimefd();
        for(int i = 0; i < times; i++) {
            RunTimerTask();
        }
    }
    void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc& cb) {
        SharedTask st(new TimeTask(id, delay, cb));
        st->SetRelease(std::bind(&TimeWheel::RemoveTimer, this, id));
        int pos = (_tick + delay) % _capacity;
        _wheel[pos].push_back(st);
        _timers[id] = WeakTask(st);
    }
    void TimerRefreshInLoop(uint64_t id) {
        auto it = _timers.find(id);
        if(it == _timers.end()) {
            return;
        }
        SharedTask st = it->second.lock();
        int delay = st->DelayTime();
        int pos = (_tick + delay) % _capacity;
        _wheel[pos].push_back(st);
    }
    void TimerCancelInLoop(uint64_t id) {
        auto it = _timers.find(id);
        if(it == _timers.end()) {
            return;
        }
        SharedTask st = it->second.lock();
        if(st) {
            st->Cancel();
        }
    }
public:
    TimeWheel(EventLoop* loop) : _capacity(60), _tick(0), _wheel(_capacity), _loop(loop), _timerfd(CreateTimefd()), _timer_channel(new Channel(_loop, _timerfd)) {
        _timer_channel->SetReadCallback(std::bind(&TimeWheel::OnTime, this));
        _timer_channel->EnableRead();
    }
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb);
    void TimerRefresh(uint64_t id);
    void TimerCancel(uint64_t id);
    bool HasTimer(uint64_t id) {
        auto it = _timers.find(id);
        if(it == _timers.end()) {
            return false;
        }
        return true;
    }
};
// ******************************
// 这里实现的是我们的是eventloop模块
class EventLoop {
private:
    using Functor = std::function<void()>;
    std::thread::id _thread_id;
    int _event_fd; // 事件的同步机制
    std::unique_ptr<Channel> _event_channel;
    Poller _poller;
    std::vector<Functor> _tasks; // 任务池
    std::mutex _mutex;
    TimeWheel _timer_wheel; // 定时器
public:
    void RunAllTask() {
        std::vector<Functor> funtor;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(funtor);
        }
        for(auto& f : funtor) {
            f();
        }
        return;
    }
    static int CreateEventfd() {
        int efd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        if(efd < 0) {
            ERR_LOG("create evenetfd error");
            abort();
        }
        return efd;
    }
    void ReadEventfd() {
        uint64_t res = 0;
        int n = read(_event_fd, &res, sizeof(res));
        if(n < 0) {
            // 前面是被信号中断了，后面是没有数据可读了
            if(errno == EINTR || errno == EAGAIN) {
                return;
            }
            ERR_LOG("read eventfd error");
            abort();
        }
        return;
    }
    void WeakEventfd() {
        uint64_t val = 1;
        int n = write(_event_fd, &val, sizeof(val));
        if(n < 0) {
            if(errno == EINTR) {
                return;
            }
            ERR_LOG("write eventfd error");
            abort();
        }
        return;
    }
public: 
    EventLoop() 
    : _thread_id(std::this_thread::get_id()),_event_fd(CreateEventfd()), _event_channel(new Channel(this, _event_fd)), _timer_wheel(this) {
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));
        _event_channel->EnableRead();
    }
    void Start() {
        while (1) {
            std::vector<Channel*> active;
            _poller.Poll(&active);
            for(auto& ch : active) {
                ch->HandleEvent();
            }
            RunAllTask();
        }
    }
    bool IsInLoop() {
        return (_thread_id == std::this_thread::get_id());
    }
    void AssertInLoop() {
        assert(_thread_id == std::this_thread::get_id());
    }
    void RunInLoop(const Functor& cb) {
        if(IsInLoop()) {
            cb();
        }
        return QueueInLoop(cb);
    }
    void QueueInLoop(const Functor& cb) {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }
        WeakEventfd();
    }
    void UpdateEvent(Channel* channel) {
        return _poller.UpdateEvent(channel);
    }
    void RemoveEvent(Channel* channel) {
        return _poller.RemoveEvent(channel);
    }
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb) {
        return _timer_wheel.TimerAdd(id, delay, cb);
    }
    void TimerCancel(uint64_t id) {
        return _timer_wheel.TimerCancel(id);
    }
    void TimerRefresh(uint64_t id) {
        return _timer_wheel.TimerRefresh(id);
    }
    bool HasTimer(uint64_t id) {
        return _timer_wheel.HasTimer(id);
    }
};
class Any {
    private:
        class holder {
            public:
                virtual ~holder() {}
                virtual const std::type_info& type() = 0;
                virtual holder* clone() = 0;
        };
        template<class T>
        class placeholder : public holder {
            public:
                placeholder(const T& val)
                :_val(val) {}
                virtual const std::type_info& type() {
                    return typeid(T);
                }
                virtual holder* clone() {
                    return new placeholder(_val);
                }
            public:
                T _val;
        };
    private:
        holder* _content;
    public:
        Any(): _content(NULL) {}
        template<typename T>
        Any(const T& val) : _content(new placeholder<T>(val)) {}
        Any(const Any& other) : _content(other._content ? other._content->clone() : NULL) {}
        ~Any() {
            delete _content;
        }
        Any& swap(Any& other) {
            std::swap(_content, other._content);
            return *this;
        }
        template<class T>
        T* get() {
            assert(typeid(T) == _content->type());
            return &((placeholder<T>*)_content)->_val; 
            // 这里直接不返回值的几个原因：
            // 1、性能问题，可能值本身比较大不好返回（消耗大）
            // 2、安全问题，不匹配时无法安全返回了
            // 3、和库里面的保持一致
        }
        template<class T>
        Any& operator=(const T& val) {
            Any(val).swap(*this);
            return *this;
        }
        Any& operator=(const Any& other) {
            Any(other).swap(*this);
            return *this;
        }
};
class Connection;
typedef enum{ DISCONNECTED, CONNECTING, CONNECTED, DISCONNECTING }ConnStatus;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection> {
private:
    uint64_t _conn_id; // 连接的唯一ID，便于连接的管理和查找，同时为了简化操作这个ID也作为了_timer_id
    int _sockfd;
    bool _enable_inactive_release; // 连接是不是启动了不活跃就销毁的选项
    EventLoop* _loop; // 连接关联的eventloop
    ConnStatus _statu; // 连接的状态
    Socket _socket; // 套接字的操作管理
    Channel _channel; // 连接的事件管理
    Buffer _in_buffer; // 输入缓冲区
    Buffer _out_buffer; // 输出缓冲区
    Any _context; // 请求的接收处理上下文数据
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer*)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
    ClosedCallback _server_closed_callback; // 这里的关闭是服务器自己的一个管理操作的回调函数
private:
    void HandleRead() {
        char buffer[65536];
        ssize_t ret = _socket.Recv(buffer, 65535);
        if(ret < 0) { // 对端关闭了
            return ShutdownInLoop();
        }
        _in_buffer.WriteAndPush(buffer, ret);
        if(_in_buffer.ReadAbleSize() > 0) { 
            return _message_callback(shared_from_this(), &_in_buffer);
        }
    }
    void HandleWrite() {
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
        if(ret < 0) {
            if(_in_buffer.ReadAbleSize() > 0) {
                _message_callback(shared_from_this(), &_in_buffer);
            }
            return Release();
        }
        _out_buffer.MoveReadOffset(ret);
        if(_out_buffer.ReadAbleSize() == 0) {
            _channel.DisableWrite();
            if(_statu == DISCONNECTING) {
                return Release();
            }
        }
    }
    void HandleClose() {
        if(_in_buffer.ReadAbleSize() > 0) {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        return Release();
    }
    void HandleError() {
        return HandleClose();
    }
    void HandleEvent() {
        if(_enable_inactive_release == true) {
            _loop->TimerRefresh(_conn_id);
        }
        if(_event_callback) {
            _event_callback(shared_from_this());
        }
    }
    void EstablishedInLoop() {
        assert(_statu == CONNECTING); // 当前的状态必须是上层的半连接状态
        _statu = CONNECTED;
        _channel.EnableRead();
        if(_connected_callback) {
            _connected_callback(shared_from_this());
        }
    }
    // 实际释放的接口
    void ReleaseInLoop() {
        _statu = DISCONNECTED;
        _channel.Remove();
        _socket.Close();
        if(_loop->HasTimer(_conn_id)) {
            CancelInactiveReleaseInLoop();
        }
        if(_closed_callback) {
            _closed_callback(shared_from_this());
        }
        if(_server_closed_callback) {
            _server_closed_callback(shared_from_this());
        }
    }
    void SendInLoop(Buffer& buf) {
        if(_statu == DISCONNECTED) {
            return;
        }
        _out_buffer.WriteBufferAndPush(buf);
        if(_channel.WriteAble() == false) _channel.EnableWrite();
    }
    void ShutdownInLoop() {
        _statu = DISCONNECTING;
        if(_in_buffer.ReadAbleSize() > 0) {
            if(_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        if(_out_buffer.ReadAbleSize() > 0) {
            if(_channel.WriteAble() == false) {
                _channel.EnableWrite();
            }
        }
        if(_out_buffer.ReadAbleSize() == 0) {
            Release();
        }
    }
    // 启动非活跃连接的超时释放规则
    void EnableInactiveReleaseInLoop(int sec) {
        _enable_inactive_release = true;
        if(_loop->HasTimer(_conn_id)) { // 已经有了定时销毁任务，我们就刷新一下
            return _loop->TimerRefresh(_conn_id);
        }
        _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
    }
    // 取消定时销毁任务
    void CancelInactiveReleaseInLoop() {
        _enable_inactive_release = false;
        if(_loop->HasTimer(_conn_id)) {
            _loop->TimerCancel(_conn_id);
        }
    }
    void UpgredeInLoop(const Any& context, 
        const ConnectedCallback& conn, 
        const MessageCallback& mesg, 
        const ClosedCallback& closed, 
        const AnyEventCallback& event) {
        _context = context;
        _connected_callback = conn;
        _message_callback = mesg;
        _closed_callback = closed;
        _event_callback = event;
    }
public:
    Connection(EventLoop* loop, uint64_t conn_id, int sockfd) : _conn_id(conn_id), _sockfd(sockfd), _enable_inactive_release(false),
    _loop(loop), _statu(CONNECTING), _channel(loop, _sockfd){
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleError, this));
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
    }
    ~Connection() {
        DBG_LOG("release connection : %p", this);
    }
    int Fd() {
        return _sockfd;
    }
    int Id() {
        return _conn_id;
    }
    bool Connected() {
        return _statu == CONNECTED;
    }
    void SetContext(const Any& context) {
        _context = context;
    }
    Any* GetContext() {
        return &_context;
    }
    void SetConnectedCallback(const ConnectedCallback& cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback& cb) { _message_callback = cb; }
    void SetCloseCallback(const ClosedCallback& cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback& cb) { _event_callback = cb; }
    void SetSrvClosedCallback(const ClosedCallback& cb) { _server_closed_callback = cb; }
    void Established() {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this)); 
    }
    void Send(const char* data, size_t len) {
        Buffer buf;
        buf.WriteAndPush(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf))); // 这里最好是写成move，因为这里的bind本质是传值传参（内部重新构造），所以
        // 传入了右值触发移动构造，然后再由bind内部的buffer穿引用给函数sendinloop
    }
    void Shutdown() {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }
    void Release() {
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }
    void EnableInactiveRelease(int sec) {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    void CancelInactiveRelease() {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    // 这里是切换协议操作，重置上下文和回调函数，必须在eventloop线程中去执行才可以，并且需要保留上下文数据才可以，反之切换的任务没有被执行
    void Upgrede(const Any& context, const ConnectedCallback& conn, const MessageCallback& msg, const ClosedCallback& closed, const AnyEventCallback& event) {
        _loop->AssertInLoop(); // 断言是不是当前需要执行的线程 
        _loop->RunInLoop(std::bind(&Connection::UpgredeInLoop, this, context, conn, msg, closed, event));
    }
};
class Acceptor {
private:
    Socket _socket;
    EventLoop* _loop;
    Channel _channel;
    using AcceptCallback = std::function<void(int)>;
    AcceptCallback _accept_callback;
private:
    void HandleRead() {
        int newfd = _socket.Accept();
        if(newfd < 0) {
            return;
        }
        if(_accept_callback) { _accept_callback(newfd); }
    }
    int CreateServer(int port) {
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.Fd();
    }
public:
    Acceptor(EventLoop* loop, int port) : _socket(CreateServer(port)), _loop(loop), _channel(loop, _socket.Fd()) {
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }
    void SetAcceptCallback(const AcceptCallback& cb) {
        _accept_callback = cb;
    }
    void Listen() {
        _channel.EnableRead();
    }
};
void Channel::Remove() { 
    return _loop->RemoveEvent(this); 
}
void Channel::Update() { 
    return _loop->UpdateEvent(this); 
}
void TimeWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb) {
    _loop->RunInLoop(std::bind(&TimeWheel::TimerAddInLoop, this, id, delay, cb));
}
void TimeWheel::TimerRefresh(uint64_t id) {
    _loop->RunInLoop(std::bind(&TimeWheel::TimerRefreshInLoop, this, id));
}
void TimeWheel::TimerCancel(uint64_t id) {
    _loop->RunInLoop(std::bind(&TimeWheel::TimerCancelInLoop, this, id));
}